package org.code.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.code.common.Constant;
import org.code.util.ByteStringConverter;
import org.code.util.DateUtil;
import org.code.util.ThreadUtil;
import org.junit.Test;

import java.util.List;

/**
 * @author xiao
 * <p>
 * 发送延迟消息
 */
public class Delay {

    static final String TOPIC = "delay-topic";

    @Test
    public void delayProducer() throws Exception {
        // 创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("delay-producer-group");
        // 设置NameServer地址
        producer.setNamesrvAddr(Constant.NAME_SERVER);
        // 启动生产者
        producer.start();
        // 创建消息
        for (int i = 0; i < Constant.SEND_MESSAGE_TIMES; i++) {
            Message message = new Message(TOPIC, ByteStringConverter.stringToByteArray("delay-message-" + i));
            // 设置延迟时间 10s
            message.setDelayTimeLevel(3);
            // 发送消息
            producer.send(message);
            System.out.printf("消息发送成功! time : %s\n", DateUtil.getNowTime());
        }
        // 关闭生产者
        producer.shutdown();
    }

    @Test
    public void delayConsumer() throws Exception {
        // 创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("delay-consumer-group");
        // 设置NameServer地址
        consumer.setNamesrvAddr(Constant.NAME_SERVER);
        // 订阅主题
        consumer.subscribe(TOPIC, "*");
        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                for (MessageExt messageExt : list) {
                    System.out.printf("消息接收成功! time : %s, msgId : %s, msgBody : %s\n",
                            DateUtil.getNowTime(),
                            messageExt.getMsgId(),
                            ByteStringConverter.byteArrayToString(messageExt.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者
        consumer.start();
        // 阻塞
        ThreadUtil.read();
    }
}
